/**
 * Copyright (C) @2014 Webank Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.webank.framework.seq.biz.service.impl;

import java.sql.SQLException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import cn.webank.framework.exception.SysException;
import cn.webank.framework.seq.biz.service.SeqService;
import cn.webank.framework.seq.integration.dao.SeqDao;
import cn.webank.framework.seq.model.Sequence;
import cn.webank.framework.seq.model.SequenceRange;
import cn.webank.framework.utils.WeBankContextUtil;

/**
 * @author jonyang
 *
 */
@Service("cn.webank.framework.seq.biz.service.SeqService")
public class SeqPojoService implements SeqService {

	private static final Logger LOG = LoggerFactory.getLogger(SeqPojoService.class);

	protected static final long DELTA = 100000000L;

	@Autowired
	@Qualifier("cn.webank.framework.seq.integration.dao.SeqDao")
	private SeqDao seqDao;

	// private volatile SequenceRange currentRange;
	private ConcurrentHashMap<String, SequenceRange> currentRangeMap = new ConcurrentHashMap<String, SequenceRange>();

	private static Lock configLock = new ReentrantLock();
	private static final int DEFAULT_STEP = 2000;
	private static final int DEFAULT_RETRYTIME = 100;
	private static int step = DEFAULT_STEP;
	private static int retryTime = DEFAULT_RETRYTIME;

	public static void setRetryTime(int aRetryTime) {
		retryTime = aRetryTime;
	}

	public SeqDao getSeqDao() {
		return seqDao;
	}

	public void setSeqDao(SeqDao aSeqDao) {
		seqDao = aSeqDao;
	}

	public static int getStep() {
		return step;
	}

	public static void setStep(int aStep) {
		step = aStep;
	}

	@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = { Exception.class }, timeout = 10)
	@Override
	public String nextValue(String moduleId, Date currentDate) {
		Assert.notNull(moduleId, "moduleId must not be null");
		long value = -1L;
		Date currentTime = currentDate;
		SequenceRange currentRange = this.currentRangeMap.get(moduleId);
		if ((currentRange == null) || isNeedReset(currentRange, currentTime)) {
			configLock.lock();
			try {
				if (currentRange == null || isNeedReset(currentRange, currentTime)) {
					currentRange = nextRange(moduleId, currentTime);
					// if ((currentRange == null)
					// || (nextRange.compareTo(currentRange) > 0)) {
					// currentRange = nextRange;
					currentRangeMap.put(moduleId, currentRange);
					// } else {
					// throw new SysException("nextRange("
					// + nextRange.getMin() + "," + nextRange.getMax()
					// + ") is low than currentRange("
					// + currentRange.getMin() + ","
					// + currentRange.getMax() + ")");
					// }
				}

			} finally {
				configLock.unlock();
			}
		}

		value = currentRange.incrementAndGet();

		if (value == -1L) {
			configLock.lock();
			try {
				currentRange = nextRange(moduleId, currentTime);
				// currentRange = nextRange(moduleId, currentTime);

				// if ((currentRange == null)
				// || (nextRange.compareTo(currentRange) > 0)) {
				// currentRange = nextRange;
				currentRangeMap.put(moduleId, currentRange);
				// } else {
				// throw new SysException("nextRange(" + nextRange.getMin()
				// + "," + nextRange.getMax()
				// + ") is low than currentRange("
				// + currentRange.getMin() + ","
				// + currentRange.getMax() + ")");
				// }
			} finally {
				configLock.unlock();
			}
			value = currentRange.incrementAndGet();
		}

		if (value < 0L) {
			throw new SysException("Sequence value overflow,value=" + value);
		}

		return "" + value;
	}

	private boolean isNeedReset(Sequence seq, Date modifiedTime) {
		boolean result = false;
		Date currentDate = modifiedTime;
		Date lastModifiedDate = seq.getModifiedTime();
		if ((currentDate.after(lastModifiedDate)) && (!DateFormatUtils.format(currentDate, "yyyy-MM-dd")
				.equals(DateFormatUtils.format(lastModifiedDate, "yyyy-MM-dd")))) {// 每天重置为0
			result = true;
		}
		return result;
	}

	private boolean isNeedReset(SequenceRange seq, Date modifiedTime) {
		boolean result = false;
		Date currentDate = modifiedTime;
		Date lastModifiedDate = seq.getModifiedTime();

		if ((currentDate.after(lastModifiedDate)) && (!(DateFormatUtils.format(currentDate, "yyyy-MM-dd"))
				.equals(DateFormatUtils.format(lastModifiedDate, "yyyy-MM-dd")))) {// 每天重置为0
			result = true;
		}
		return result;
	}

	private SequenceRange nextRange(String moduleId, Date aCurrentTime) {
		Date currentTime = aCurrentTime;
		for (int i = 0; i < retryTime; i++) {
			long oldValue = -1L;
			Sequence seq = null;
			try {
				seq = queryOldValueByModuleId(moduleId);

				if (seq == null) {
					throw new SysException("Sequence isn't define. moduleId:" + moduleId);
				}

				oldValue = seq.getCurrentValue();

				boolean resetFlag = isNeedReset(seq, currentTime);
				if ((resetFlag) && (updateNewValue(moduleId, oldValue, 0L, currentTime) < 1)) {
					continue;
				}

				if (resetFlag) {
					oldValue = 0L;
				}
				// else {
				// oldValue = seq.getCurrentValue();
				// }
			} catch (SQLException e) {
				throw new SysException("faild to get seqence value", e);
			}

			if (!isOldValueFixed(oldValue, moduleId)) {
				throw new SysException("old value:" + oldValue + "  is invalid");
			}

			long newValue = generateNewValue(oldValue, step);
			try {
				if ((seq != null) && (seq.getModifiedTime().after(currentTime))) {
					currentTime = seq.getModifiedTime();
				}

				if (updateNewValue(moduleId, oldValue, newValue, currentTime) > 0) {
					// return new SequenceRange(oldValue + 1L, oldValue + step,
					// currentTime);
					return new SequenceRange(oldValue, oldValue + step, currentTime);
				}
			} catch (SQLException e) {
				throw new SysException("faild to get seqence value", e);
			}
		}
		return (SequenceRange) null;
	}

	private boolean isOldValueFixed(long oldValue, String moduleId) {
		boolean result = true;

		StringBuilder msg = new StringBuilder();
		if (oldValue < 0L) {
			msg.append("Sequence value cann't be less than zero.");
			result = false;
		} else if (oldValue > 9223372036754775807L) {
			msg.append("Sequence value overflow.");
			result = false;
		}

		if (!result) {
			msg.append(" Sequence value = ").append(oldValue);
			msg.append(" , check moduleId = ").append(moduleId);
			LOG.warn(msg.toString());
		}
		return result;
	}

	private Sequence queryOldValueByModuleId(String moduleId) throws SQLException {
		return seqDao.queryOldValueByModuleId(moduleId);
	}

	private long generateNewValue(long oldValue, int step) {
		long newValue = oldValue + step;
		return newValue;
	}

	private int updateNewValue(String moduleId, long oldValue, long newValue, Date modifiedTime) throws SQLException {
		Map<String, Object> parameterMap = new HashMap<String, Object>();
		parameterMap.put("moduleId", moduleId);
		parameterMap.put("oldValue", Long.valueOf(oldValue));
		parameterMap.put("newValue", Long.valueOf(newValue));
		parameterMap.put("modifiedTime", modifiedTime);
		return seqDao.updateNewValue(parameterMap);
	}

	private String nextSeqNo(String moduleId, String seqType, String dcnNo, Date currentDate) {
		String seq = nextValue(moduleId, currentDate);
		String date = DateFormatUtils.format(currentDate, "yyMMdd");
		String seqNo = date + seqType + dcnNo + moduleId + org.apache.commons.lang3.StringUtils.leftPad(seq, 18, '0');
		return seqNo;
	}

	public String nextBizSeqNo(String moduleId, String dcnNo, Date currentDate) {
		String bizSeqNo = WeBankContextUtil.getBizSeqNo();
		if (!StringUtils.hasLength(bizSeqNo)) {
			bizSeqNo = this.nextSeqNo(moduleId, BIZ_SEQ, dcnNo, currentDate);
			WeBankContextUtil.setBizSeqNo(bizSeqNo);
		}
		return bizSeqNo;

	}

	public String nextConsumerSeqNo(String moduleId, String dcnNo, Date currentDate) {
		return this.nextSeqNo(moduleId, CONSUMER_SEQ, dcnNo, currentDate);
	}

	@Override
	public String nextSeqNo(String moduleId, String dcnNo, char type, Date currentDate) {
		return this.nextSeqNo(moduleId, String.valueOf(type), dcnNo, currentDate);
	}

}
